package com.cdtye.shuoh.service.device.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.cdtye.common.webcore.web.AjaxJson;
import com.cdtye.shuoh.config.fastdfs.FastDfsUtils;
import com.cdtye.shuoh.config.mqtt.MqttPushClient;
import com.cdtye.shuoh.config.redis.RedisService;
import com.cdtye.shuoh.config.websocket.event.MessageEvent;
import com.cdtye.shuoh.domain.Frame.AtmosphereFrame;
import com.cdtye.shuoh.domain.Frame.IceCoverFrame;
import com.cdtye.shuoh.domain.Frame.ImageDataFrame;
import com.cdtye.shuoh.domain.Frame.ImageRequestEndFrame;
import com.cdtye.shuoh.domain.constant.CacheConstants;
import com.cdtye.shuoh.domain.device.entity.Device;
import com.cdtye.shuoh.domain.device.entity.DeviceDatas;
import com.cdtye.shuoh.domain.device.entity.DeviceInfo;
import com.cdtye.shuoh.domain.device.model.DeviceInfos;
import com.cdtye.shuoh.domain.device.model.TlvDeviceParams;
import com.cdtye.shuoh.domain.device.model.TopoAddDatas;
import com.cdtye.shuoh.domain.enums.*;
import com.cdtye.shuoh.domain.product.entity.Product;
import com.cdtye.shuoh.domain.product.entity.ProductServices;
import com.cdtye.shuoh.domain.tdengine.Fields;
import com.cdtye.shuoh.domain.tdengine.SuperTableDto;
import com.cdtye.shuoh.domain.tdengine.TableDto;
import com.cdtye.shuoh.mapper.device.DeviceDatasMapper;
import com.cdtye.shuoh.service.device.DeviceDatasService;
import com.cdtye.shuoh.service.device.DeviceInfoService;
import com.cdtye.shuoh.service.product.ProductService;
import com.cdtye.shuoh.service.product.ProductServicesService;
import com.cdtye.shuoh.service.product.ProductTemplateService;
import com.cdtye.shuoh.service.rule.RuleService;
import com.cdtye.shuoh.service.tgengine.TdEngineService;
import com.cdtye.shuoh.util.text.*;
import com.cdtye.shuoh.util.text.UUID;
import com.google.common.base.CaseFormat;
import com.jayway.jsonpath.JsonPath;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class DeviceDatasServiceImpl extends ServiceImpl<DeviceDatasMapper, DeviceDatas> implements DeviceDatasService {

    @Autowired
    private DeviceServiceImpl deviceService;

    @Autowired
    private DeviceInfoService deviceInfoService;

    @Autowired
    private ProductService productService;

    @Autowired
    private ProductServicesService productServicesService;

    @Autowired
    private TdEngineService tdEngineService;


    @Autowired
    private DeviceDatasMapper deviceDatasMapper;

    @Autowired
    private ProductTemplateService productTemplateService;

    @Autowired
    private RedisService redisService;

    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    @Autowired
    private RuleService ruleService;

    @Autowired
    private FastDfsUtils fastDfsUtils;

    @Autowired
    private MqttPushClient mqttPushClient;

    /**
     * 数据库名称
     */
    @Value("${spring.datasource.dynamic.datasource.tdengine.dbName:cms_shuoh}")
    String dataBaseName;

    @Value("${fastdfs.web_url}")
    private String fdfsPrefix;

    /**
     * 数据解析
     * @param topic
     * @param message
     * @throws Exception
     * @Description 使用final 保证线程安全
     *
     *
     * Topic	                        Publisher（发布者）  Subscriber（订阅者）	用途
     * /v1/devices/{deviceId}/topo/add	边设备	物联网平台	边设备添加子设备
     * /v1/devices/{deviceId}/topo/addResponse	物联网平台	边设备	物联网平台返回的添加子设备的响应
     * /v1/devices/{deviceId}/topo/delete	边设备	物联网平台	边设备删除子设备
     * /v1/devices/{deviceId}/topo/deleteResponse	物联网平台	边设备	物联网平台返回的删除子设备的响应
     * /v1/devices/{deviceId}/topo/update	边设备	物联网平台	边设备更新子设备状态
     * /v1/devices/{deviceId}/topo/updateResponse	物联网平台	边设备	物联网平台返回的更新子设备状态的响应
     * /v1/devices/{deviceId}/datas	边设备	物联网平台	边设备上报数据
     * /v1/devices/{deviceId}/command	物联网平台	边设备	物联网平台给设备或边设备下发命令
     * /v1/devices/{deviceId}/commandResponse	边设备	物联网平台	边设备返回给物联网平台的命令响应
     *
     */
    @Async("linkAsync")
    public void insertBaseDatas(String topic, MqttMessage message) throws Exception {
        DeviceDatas deviceDatas = new DeviceDatas();
        String result = new String(message.getPayload(),"UTF-8");
        deviceDatas.setMessageId(String.valueOf(message.getId()));
        deviceDatas.setTopic(topic);
        deviceDatas.setQos(message.getQos());
        deviceDatas.setMessage(result);

        if (topic.startsWith("/v1/devices/") && topic.endsWith("/topo/add")){
            String deviceIdentification = StringUtils.substringBetween(topic,"/v1/devices/","/topo/add");
            deviceDatas.setDeviceIdentification(deviceIdentification);
            String payload = this.processingTopoAddTopic(deviceIdentification, result);
            String responseTopic = topic.replace("add", "addResponse");
            mqttPushClient.publish(responseTopic,payload);
        }else if (topic.startsWith("/v1/devices/") && topic.endsWith("/topo/delete")) {
            String deviceIdentification = StringUtils.substringBetween(topic,"/v1/devices/","/topo/delete");
            deviceDatas.setDeviceIdentification(deviceIdentification);
            final String payload = this.processingTopoDeleteTopic(deviceIdentification, result);
            final Map<String, Object> param = new HashMap<>();
            String responseTopic = topic.replace("delete", "deleteResponse");
            mqttPushClient.publish(responseTopic,payload);
        } else if (topic.startsWith("/v1/devices/") && topic.endsWith("/topo/update")) {
            String deviceIdentification = StringUtils.substringBetween(topic,"/v1/devices/","/topo/update");
            deviceDatas.setDeviceIdentification(deviceIdentification);
            final String payload = this.processingTopoUpdateTopic(deviceIdentification, result);
            String responseTopic = topic.replace("update", "updateResponse");
            mqttPushClient.publish(responseTopic,payload);
        } else if (topic.startsWith("/v1/devices/") && topic.endsWith("/datas")) {
            String deviceIdentification = StringUtils.substringBetween(topic,"/v1/devices/","/datas");
            deviceDatas.setDeviceIdentification(deviceIdentification);
            this.processingDatasTopic(deviceIdentification, result);
        } else if (topic.startsWith("/v1/devices/") && topic.endsWith("/commandResponse")) {
            //this.processingTopoCommandResponseTopic(deviceIdentification, body);
        } else if(topic.startsWith("$SYS/brokers/") && (topic.endsWith("/connected") || topic.endsWith("/disconnected"))){
            String[] str = topic.split("/");
            String clientId = str[str.length-2];
            if(topic.endsWith("/connected")){
                deviceService.updateDeviceConnectStatusByClientId(clientId,DeviceConnectStatus.ONLINE.getValue());
            }else if(topic.endsWith("/disconnected")){
                deviceService.updateDeviceConnectStatusByClientId(clientId,DeviceConnectStatus.OFFLINE.getValue());
            }
        }else {
            List<Device> deviceList = deviceService.findByDeviceType(DeviceType.LORA_4G_GATWAY.getValue());
            deviceList.forEach(device -> {
                if(device.getDeviceIdentification().equals(StringUtils.remove(topic,"/"))){
                    try {
                        processingTLVDatasTopic(result,device.getDeviceIdentification(),device.getId());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }

        deviceDatasMapper.insert(deviceDatas);

    }

    /**
     * MQTT网关设备增加子设备
     * @param deviceIdentification
     * @param result
     * @return
     * @throws Exception
     */
    public String processingTopoAddTopic(String deviceIdentification, String result) throws Exception {
        TopoAddDatas topoAddDatas = JSON.toJavaObject(JSON.parseObject(result), TopoAddDatas.class);
        Map<Object, Object> responseMaps = new HashMap<>();
        List<Map<String, Object>> dataList = new ArrayList();
        responseMaps.put("mid", topoAddDatas.getMid());
        responseMaps.put("statusCode", 0);
        responseMaps.put("statusDesc", "Success!");
        responseMaps.put("data", dataList);
        //找到子设备的网关设备
        Device device = deviceService.findOneByDeviceIdentification(deviceIdentification);
        if (null == device) {
            log.error("The side device reports data processing, but the device does not exist,DeviceIdentification:{},Result:{}" + deviceIdentification + result);
            responseMaps.put("statusCode", 1);
            responseMaps.put("statusDesc", "The side device reports data processing, but the device does not exist.");
            return JSON.toJSONString(responseMaps);
        }
        for (DeviceInfos deviceInfos : topoAddDatas.getDeviceInfos()) {
            Product product = productService.findOneByModel(deviceInfos.getModel());
            Map responseMap = new HashMap<>();
            if (null == product) {
                log.error("The side device reports data processing, but the product does not exist,DeviceIdentification:{},Result:{}" + deviceInfos.getModel() + result);
                responseMap.put("statusCode", 1);
                responseMap.put("statusDesc", "The side device reports data processing, but the product does not exist.");
                Map<Object, Object> deviceInfoMap = new HashMap<>();
                deviceInfoMap.put("nodeId", deviceInfos.getNodeId());
                deviceInfoMap.put("name", deviceInfos.getName());
                deviceInfoMap.put("description", deviceInfos.getDescription());
                deviceInfoMap.put("manufacturerId", deviceInfos.getManufacturerId());
                deviceInfoMap.put("model", deviceInfos.getModel());
                responseMap.put("deviceInfo", deviceInfoMap);
                dataList.add(responseMap);
            }else {
                if(null==deviceInfos.getNodeId()){
                    log.error("The side device reports data processing, but the nodeId of the deviceInfo is null");
                    responseMap.put("statusCode", 1);
                    responseMap.put("statusDesc", "The side device reports data processing, but the nodeId of the deviceInfo is null.");
                    Map<Object, Object> deviceInfoMap = new HashMap<>();
                    deviceInfoMap.put("nodeId", deviceInfos.getNodeId());
                    deviceInfoMap.put("name", deviceInfos.getName());
                    deviceInfoMap.put("description", deviceInfos.getDescription());
                    deviceInfoMap.put("manufacturerId", deviceInfos.getManufacturerId());
                    deviceInfoMap.put("model", deviceInfos.getModel());
                    responseMap.put("deviceInfo", deviceInfoMap);
                    dataList.add(responseMap);
                }else{
                    DeviceInfo deviceInfo = new DeviceInfo();
                    deviceInfo.setDId(device.getId());
                    deviceInfo.setNodeId(deviceInfos.getNodeId());
                    deviceInfo.setNodeName(deviceInfos.getName());
                    deviceInfo.setDescription(deviceInfos.getDescription());
                    deviceInfo.setManufacturerId(deviceInfos.getManufacturerId());
                    deviceInfo.setModel(deviceInfos.getModel());
                    deviceInfo.setConnectStatus(DeviceConnectStatus.INIT.getValue());
                    DeviceInfo record = deviceInfoService.insertOrUpdateByNodeId(deviceInfo);
                    Map<Object, Object> deviceInfoMap = new HashMap<>();
                    deviceInfoMap.put("deviceId", record.getDeviceId());
                    deviceInfoMap.put("nodeId", record.getNodeId());
                    deviceInfoMap.put("name", record.getNodeName());
                    deviceInfoMap.put("description", record.getDescription());
                    deviceInfoMap.put("manufacturerId", record.getManufacturerId());
                    deviceInfoMap.put("model", record.getModel());
                    responseMap.put("deviceInfo", deviceInfoMap);
                    //创建TD普通表
                    List<ProductServices> allByProductId = productServicesService.findAllByProductIdentificationId(product.getProductIdentification());
                    TableDto tableDto;
                    for (ProductServices productServices : allByProductId) {
                        tableDto = new TableDto();
                        tableDto.setDataBaseName(dataBaseName);
                        //超级表命名规则 : 产品类型_产品标识_服务名称
                        String superTableName = TdUtils.getSuperTableName(product.getProductType(), product.getProductIdentification(), productServices.getServiceName());
                        tableDto.setSuperTableName(superTableName);
                        //子表命名规则 : 产品类型_产品标识_服务名称_设备标识（设备唯一标识）
                        tableDto.setTableName(TdUtils.getSubTableName(superTableName, deviceInfo.getDeviceId()));
                        //Tag的处理
                        List<Fields> tagsFieldValues = new ArrayList<>();
                        Fields fields = new Fields();
                        fields.setFieldValue(device.getDeviceIdentification());
                        tagsFieldValues.add(fields);
                        tableDto.setTagsFieldValues(tagsFieldValues);
                        AjaxJson ctResult = tdEngineService.createTable(tableDto);
                        if (ctResult.getStatus() == 200) {
                            log.info("Create SuperTable Success: " + ctResult.getMessage());
                        } else {
                            log.error("Create SuperTable Exception: " + ctResult.getMessage());
                        }
                    }
                    responseMap.put("statusCode", 0);
                    responseMap.put("statusDesc", "SUCCESS");
                    dataList.add(responseMap);
                }
            }
        }
        return JSON.toJSONString(responseMaps);
    }


    /**
     * MQTT网关设备删除子设备
     * @param deviceIdentification
     * @param body
     * @return
     * @throws Exception
     */
    public String processingTopoDeleteTopic(String deviceIdentification, String body) throws Exception {
        Map<Object, Object> responseMaps = new HashMap<>();
        List<Map<String, Object>> dataList = new ArrayList();
        responseMaps.put("mid", JsonPath.read(body, "$.mid").toString());
        responseMaps.put("statusCode", 0);
        responseMaps.put("statusDesc", "Success!");
        responseMaps.put("data", dataList);
        final List<String> deviceIds = JsonPath.read(body, "$.deviceIds[*]");
        if (null == deviceIds) {
            log.error("The side device reports data processing, but the deviceId does not exist,DeviceIdentification:{},Body:{}", deviceIdentification, body);
            responseMaps.put("statusCode", 1);
            responseMaps.put("statusDesc", "The side device reports data processing, but the deviceId does not exist.");
            return JSON.toJSONString(responseMaps);
        }
        for (String deviceId : deviceIds) {
            final int deleteByDeviceIdCount = deviceInfoService.deleteByDeviceId(deviceId);
            Map responseMap = new HashMap<>();
            if (deleteByDeviceIdCount > 0) {
                responseMap.put("statusCode", 0);
                responseMap.put("statusDesc", "SUCCESS");
            } else {
                responseMap.put("statusCode", 1);
                responseMap.put("statusDesc", "abortive");
                log.error("Delete DeviceInfo Exception");
            }
            responseMap.put("deviceId", deviceId);
            dataList.add(responseMap);
        }
        return JSON.toJSONString(responseMaps);
    }

    /**
     * MQTT网关设备更新子设备信息
     * @param deviceIdentification
     * @param body
     * @return
     * @throws Exception
     */
    public String processingTopoUpdateTopic(String deviceIdentification, String body) throws Exception {
        JSONObject message = JSON.parseObject(body);
        Map<Object, Object> responseMaps = new HashMap<>();
        List<Map<String, Object>> dataList = new ArrayList();
        responseMaps.put("mid", JsonPath.read(body, "$.mid").toString());
        responseMaps.put("statusCode", 0);
        responseMaps.put("statusDesc", "Success!");
        responseMaps.put("data", dataList);
        JSONArray jsonArray = message.getJSONArray("deviceStatuses");
        if (null == jsonArray) {
            log.error("The side device reports data processing, but the deviceStatus does not exist,DeviceIdentification:{},Body:{}", deviceIdentification, body);
            responseMaps.put("statusCode", 1);
            responseMaps.put("statusDesc", "The side device reports data processing, but the deviceStatus does not exist.");
            return JSON.toJSONString(responseMaps);
        }
        for (Object json : jsonArray) {
            Map responseMap = new HashMap<>();
            Map deviceStatusMap = (Map) json;
            final String deviceId = deviceStatusMap.get("deviceId").toString();
            final String status = deviceStatusMap.get("status").toString();
            final DeviceInfo deviceInfo = deviceInfoService.findOneByDeviceId(deviceId);
            if (null == deviceInfo) {
                log.error("The side device reports data processing, but the device does not exist,DeviceIdentification:{},Body:{}", deviceIdentification, body);
                responseMap.put("statusCode", 1);
                responseMap.put("statusDesc", "The side device reports data processing, but the device does not exist,DeviceIdentification");
                responseMap.put("deviceId", deviceId);
                dataList.add(responseMap);
                continue;
            }
            if ("ONLINE".equals(status)) {
                deviceInfo.setConnectStatus(DeviceConnectStatus.ONLINE.getValue());
            } else if ("OFFLINE".equals(status)) {
                deviceInfo.setConnectStatus(DeviceConnectStatus.OFFLINE.getValue());
            }
            final int updateDeviceInfoCount = deviceInfoService.updateDeviceInfo(deviceInfo);
            if (updateDeviceInfoCount > 0) {
                responseMap.put("statusCode", 0);
                responseMap.put("statusDesc", "SUCCESS");
            } else {
                responseMap.put("statusCode", 1);
                responseMap.put("statusDesc", "ABORTIVE");
                log.error("Update DeviceInfo Exception");
            }
            responseMap.put("deviceId", deviceId);
            dataList.add(responseMap);

        }
        return JSON.toJSONString(responseMaps);
    }


    /**
     * MQTT设备数据上报
     * @param deviceIdentification
     * @param body
     * @throws Exception
     */
    public void processingDatasTopic(String deviceIdentification, String body) throws Exception {
        //根据返回的json解析出上报的数据data，所属的服务serviceName，事件发生的时间eventTime
        Map<String, Object> resultMap = (Map) JSON.parse(body);
        List<Map<String, Object>> items = (List<Map<String, Object>>) resultMap.get("devices");
        for (Map<String, Object> item : items) {
            final Object deviceId = item.get("deviceId");
            Device device = deviceService.findOneByDeviceIdentification(deviceIdentification);
            if (null== device) {
                log.error("The side device reports data processing, but the device does not exist,DeviceIdentification:{},Body:{}", deviceIdentification, body);
                continue;
            }
            String manufacturerId = "";
            String model = "";
            if (device.getDeviceType().equals(DeviceType.GATEWAY.getValue()) || device.getDeviceType().equals(DeviceType.LORA_4G_GATWAY.getValue())) {
                final DeviceInfo oneByDeviceId = deviceInfoService.findOneByDeviceId(deviceId.toString());
                if (null==oneByDeviceId) {
                    log.error("The side device reports data processing, but the device does not exist,DeviceIdentification:{},Body:{}", deviceIdentification, body);
                    continue;
                }
                model = oneByDeviceId.getModel();
            } else if (device.getDeviceType().equals(DeviceType.COMMON.getValue())) {
                Product product = productService.getByProductIdentification(device.getProductIdentification());
                model = product.getModel();
            }
            final JSONArray services = JSON.parseArray(item.get("services").toString());
            //如果设备上报的数据为空，不需要存，跳过该循环，进入下个循环
            if (services.isEmpty()) {
                log.error("The side device reports data processing, but the data does not exist,DeviceIdentification:{},Body:{}", deviceIdentification, body);
                continue;
            }
            for (Object service : services) {
                final JSONObject serviceData = JSON.parseObject(service.toString());
                final Object serviceId = serviceData.get("serviceId");
                final Object eventTime = serviceData.get("eventTime");
                Map<String, Object> data = (Map) JSON.parse(serviceData.get("data").toString());
                data = CaseFormatUtils.lowerCamelToLowerUnderscore(data);
                final Product product = productService.findOneByModel( model);
                //超级表命名规则 : 产品类型_产品标识_服务名称
                String superTableName = TdUtils.getSuperTableName(product.getProductType(),product.getProductIdentification(),serviceId.toString());
                //子表命名规则 : 产品类型_产品标识_服务名称_设备标识（设备唯一标识）
                String tableName = TdUtils.getSubTableName(superTableName,deviceId.toString());
                final String TdengineJson = productTemplateService.getTdengineJsonByProductIdAndServiceId(product.getId(),serviceId.toString());
                final SuperTableDto superTableDto = JSON.toJavaObject(JSON.parseObject(TdengineJson),SuperTableDto.class);
                //获取超级表的表结构信息
                final List<Fields> schemaFields = superTableDto.getSchemaFields();
                //如果表结构信息为空，跳过该循环，进入下个循环
                if (schemaFields == null) {
                    continue;
                }
                //取出事件发生时间，并格式化为long类型的毫秒时间戳
                String eventTimeStr = eventTime.toString().replace("T", "").replace("Z", "");
                long eventDataTime = DateUtils.string2MillisWithJdk8(eventTimeStr, "yyyyMMddHHmmss");
                //超级表第一个字段数据类型必须为时间戳,默认Ts为当前系统时间
                schemaFields.get(0).setFieldValue(DateUtils.millisecondStampL());
                //因为超级表的第二个字段为事件发生时间数据类型必须为时间戳，所以直接将索引为1的字段信息对象的字段值设置为eventTime
                schemaFields.get(1).setFieldValue(eventDataTime);
                //循环设置字段值
                for (Fields schemaField : schemaFields) {
                    //根据字段名称获取data里的数据（字段名称为服务的属性，数据上报的格式就是根据属性来的）
//                    String filedName = CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, schemaField.getFieldName());
                    Object fieldValue = data.get(schemaField.getFieldName());
                    //获取的值不为空才给该字段赋值
                    if (fieldValue != null) {
                        //如果字段为字符串类型，字段值的大小不超过该列宽，才给该字段赋值（如果超过列宽，sql会报错）
                        if (schemaField.getSize() != null && "nchar".equals(schemaField.getDataType().getDataType())) {
                            if (fieldValue.toString().length() <= schemaField.getSize()) {
                                if("float".equals(schemaField.getDataType().getDataType())) {
                                    schemaField.setFieldValue((float)fieldValue * schemaField.getStep());
                                }else{
                                    schemaField.setFieldValue(fieldValue);
                                }
                            }
                        } else {
                            //其他数据类型，直接赋值
                            schemaField.setFieldValue(fieldValue);
                        }
                    }
                }
                //获取超级表的标签信息
                final List<Fields> tagsFields = superTableDto.getTagsFields();
                //循环设置字段值
                for (Fields tagsField : tagsFields) {
                    //根据业务逻辑，将超级表的标签字段定为
                    // 1:设备标识：deviceIdentification
                    Object fieldValue = deviceId;
                    //获取的值不为空才给该字段赋值
                    if (fieldValue != null) {
                        //如果字段为字符串类型，字段值的大小不超过该列宽，才给该字段赋值（如果超过列宽，sql会报错）
                        if (tagsField.getSize() != null && "nchar".equals(tagsField.getDataType().getDataType())) {
                            if (fieldValue.toString().length() <= tagsField.getSize()) {
                                tagsField.setFieldValue(fieldValue);
                            }
                        } else {
                            //其他数据类型，直接赋值
                            tagsField.setFieldValue(fieldValue);
                        }
                    }
                }
                //字段信息对象集合通过stream流过滤掉没有字段值的字段对象
                List<Fields> schemaFieldsStream = schemaFields.stream().filter(fields -> fields.getFieldValue() != null).collect(Collectors.toList());
                //字段信息对象集合通过stream流过滤掉没有字段值的字段对象
                List<Fields> tagsFieldsStream = tagsFields.stream().filter(fields -> fields.getFieldValue() != null).collect(Collectors.toList());
                //如果字段值只有第一个字段的时间戳，说明上报的数据没有符合该服务的属性，不做保存操作，跳过该循环，进入下个循环
                if (schemaFieldsStream.size() == 1) {
                    continue;
                }
                //设置插入所需参数
                TableDto tableDto = new TableDto();
                tableDto.setDataBaseName(superTableDto.getDataBaseName());
                tableDto.setSuperTableName(superTableDto.getSuperTableName());
                tableDto.setTableName(tableName);
                tableDto.setSchemaFieldValues(schemaFieldsStream);
                tableDto.setTagsFieldValues(tagsFieldsStream);
                //调用插入方法插入数据
                final AjaxJson insertResult = this.tdEngineService.insertData(tableDto);
                if (insertResult.getStatus() == 200) {
                    log.info("DeviceIdentification: {}, Insert data result: {}", deviceIdentification, insertResult.getMessage());
                } else {
                    log.error("DeviceIdentification: {}, Insert data Exception: {}", deviceIdentification, insertResult.getMessage());
                }

                //todo 报警处理
                ruleService.processingRule(product.getProductIdentification(),
                        productServicesService.findOneByProductIdentificationAndServiceId(product.getProductIdentification(),serviceId.toString()).getId(),
                        deviceId.toString(),new Date(Long.valueOf(String.valueOf(eventDataTime))),data);


                //todo websocket 推送
                System.out.println(eventTimeStr);
                System.out.println(eventDataTime);
                System.out.println(new Date(Long.valueOf(String.valueOf(eventDataTime))));
                data.put("event_time",DateUtils.getJVMDate(String.valueOf(eventDataTime),"yyyy-MM-dd HH:mm:ss"));
                MessageEvent messageEvent = new MessageEvent(this,deviceId.toString(),data.toString());
                applicationEventPublisher.publishEvent(messageEvent);
            }
        }
    }

    /**
     * MQTT裸奔TLV方式数据处理 (TLV)
     * @param body
     * @throws Exception
     */
    public void processingTLVDatasTopic(String body,String deviceIdentification,String did) throws Exception {
        List<TlvDeviceParams> list = getDevicesFromNakeMessage(body,deviceIdentification,did);
        for (TlvDeviceParams tlvDeviceParams : list) {
            processingDatasTopic(deviceIdentification,tlvDeviceParams.getDatas());
        }

    }

    /**
     * 从TLV裸奔文本中检测设备信息 (TLV)
     * @param body
     * @return
     */
    private List<TlvDeviceParams> getDevicesFromNakeMessage(String body,String deviceIdentification,String did) throws Exception {
        List<TlvDeviceParams> result = new ArrayList<>();
        JSONObject content = JSONObject.parseObject(body);
        List<Map> rxpks = JsonPath.read(content.toJSONString(), "$..rxpk[*]");
        for (Map<String,Object> map : rxpks){
            String time = map.get("time").toString();
            String readData = map.get("pkdt").toString();
            if(null == readData){
                continue;
            }
            readData = readData.replaceAll(" ","");
            TlvDeviceParams tlvDeviceParams = new TlvDeviceParams();
            String deviceId = readData.substring(2,10);
            String deviceTypeData = readData.substring(16,22);
            String deviceTypeCode = DataConverter.parseInteger2Hex(DataConverter.parseHexStr2Integer(deviceTypeData.substring(4,6)) % 128);
            String model = TLVDeviceType.parseByValue(deviceTypeCode).getModel();
            Integer versionAndLevel = DataConverter.parseHexStr2Integer(deviceTypeData.substring(2,4));
            String version = String.valueOf(versionAndLevel / 32);
            String volt_level = String.valueOf(versionAndLevel % 32);
            //todo 新建设备信息 && 创建TD普通表
            DeviceInfo deviceInfo = new DeviceInfo();
            deviceInfo.setDId(did);
            deviceInfo.setModel(model);
            deviceInfo.setNodeName(deviceId);
            deviceInfo.setNodeId(deviceId);
            deviceInfo.setDescription(model);
            deviceInfo.setConnectStatus(DeviceConnectStatus.ONLINE.getValue());
            DeviceInfo record = deviceInfoService.insertOrUpdateByNodeId(deviceInfo);
            //创建TD普通表
            Product product = productService.findOneByModel(record.getModel());
            List<ProductServices> allByProductId = productServicesService.findAllByProductIdentificationId(product.getProductIdentification());
            TableDto tableDto;
            for (ProductServices productServices : allByProductId) {
                tableDto = new TableDto();
                tableDto.setDataBaseName(dataBaseName);
                //超级表命名规则 : 产品类型_产品标识_服务名称
                String superTableName = TdUtils.getSuperTableName(product.getProductType(), product.getProductIdentification(), productServices.getServiceName());
                tableDto.setSuperTableName(superTableName);
                //子表命名规则 : 产品类型_产品标识_服务名称_设备标识（设备唯一标识）
                tableDto.setTableName(TdUtils.getSubTableName(superTableName, record.getDeviceId()));
                //Tag的处理
                List<Fields> tagsFieldValues = new ArrayList<>();
                Fields fields = new Fields();
                fields.setFieldValue(deviceIdentification);
                tagsFieldValues.add(fields);
                tableDto.setTagsFieldValues(tagsFieldValues);
                AjaxJson ctResult = tdEngineService.createTable(tableDto);
                if (ctResult.getStatus() == 200) {
                    log.info("Create SuperTable Success: " + ctResult.getMessage());
                } else {
                    log.error("Create SuperTable Exception: " + ctResult.getMessage());
                }
            }
            tlvDeviceParams.setDatas(convertToBodyTLV(readData.substring(22,readData.length()-4),record.getDeviceId(),time,version,volt_level,TLVDeviceType.parseByValue(deviceTypeCode).getKey()));
            result.add(tlvDeviceParams);
        }
        return result;
    }

    /**
     * 协议脚本转换处理（TLV）
     * @param detectDataStr
     * @param deviceId
     * @param eventTime
     * @return
     */
    private String convertToBodyTLV(String detectDataStr,String deviceId,String eventTime,String version,String volt_level,String tlvDeviceType){
        Map<String,Object> result = new HashMap<>();
        Map<String,Object> body = new HashMap<>();
        Map<String,Object> services = new HashMap<>();
        Map<String,Object> data = tlvDataAnalysis(detectDataStr,new HashMap<>(),tlvDeviceType);
        data.put("version",version);
        data.put("volt_level",volt_level);
        services.put("data",data);
        services.put("serviceId","analog");
        services.put("eventTime",DateUtils.getTimestampDate(eventTime,"yyyyMMddHHmmss"));
        body.put("deviceId",deviceId);
        body.put("services",Arrays.asList(services));
        result.put("devices",Arrays.asList(body));
        return JSONObject.toJSONString(result);
    }


    /**
     * 获取基本设备基本数据（TLV）
     * @param detectDataStr
     * @param map
     * @return
     */
    private Map<String,Object> tlvDataAnalysis(String detectDataStr,Map<String,Object> map,String tlvDeviceType){
        if(null == detectDataStr || detectDataStr.length() <= 0){
            return map;
        }
        String dataType = detectDataStr.substring(0,2);
        if("04".equals(dataType)){
            short value = DataConverter.parseSignedHex2Short(detectDataStr.substring(2,6));
            map.put("temp",value);
            return tlvDataAnalysis(detectDataStr.substring(6),map,tlvDeviceType);
        }else if("05".equals(dataType)){
            Integer value = DataConverter.parseHexStr2Integer(detectDataStr.substring(2,4));
            map.put("humidity",value);
            return tlvDataAnalysis(detectDataStr.substring(4),map,tlvDeviceType);
        }else if("01".equals(dataType)){
            String event = detectDataStr.substring(2,4);
            short xAcceleration = DataConverter.parseSignedHex2Short(detectDataStr.substring(4,8));
            short yAcceleration = DataConverter.parseSignedHex2Short(detectDataStr.substring(8,12));
            short zAcceleration = DataConverter.parseSignedHex2Short(detectDataStr.substring(12,16));
            map.put("event",event);
            map.put("x_acc_val",xAcceleration);
            map.put("y_acc_val",yAcceleration);
            map.put("z_acc_val",zAcceleration);
            return tlvDataAnalysis(detectDataStr.substring(18),map,tlvDeviceType);
        }else if("08".equals(dataType)){
            String value = String.valueOf(DataConverter.parseHex2Float(detectDataStr.substring(2,10)));
            if(TLVDeviceType.DISTANCE.getKey().equals(tlvDeviceType)){
                map.put("rng_val",value);
            }else if(TLVDeviceType.DISPLACEMENT.getKey().equals(tlvDeviceType)){
                map.put("pos_val",value);
            }
            return tlvDataAnalysis(detectDataStr.substring(10),map,tlvDeviceType);
        }else if("11".equals(dataType)){
            Integer length = Integer.valueOf(detectDataStr.substring(2,4),16)/4;
            for(int i=1;i<=length;i++){
                if(i==1){
                    Integer xAngle = DataConverter.parseSignedHex2Integer(detectDataStr.substring(4,12));
                    map.put("ang_val",xAngle);
                }else if(i==2){
                    Integer yAngle = DataConverter.parseSignedHex2Integer(detectDataStr.substring(12,20));
                    map.put("ang_val",yAngle);
                }else if(i==3){
                    Integer zAngle = DataConverter.parseSignedHex2Integer(detectDataStr.substring(20,28));
                    map.put("ang_val",zAngle);
                }
            }
            return tlvDataAnalysis(detectDataStr.substring(length*8+4),map,tlvDeviceType);

        }
        return map;
    }


    /**
     * 处理结冰设备的气象环境类数据报
     * @param atmosphereFrame
     * @return
     */
    public String processingAtmosphereFrame(AtmosphereFrame atmosphereFrame){
        Map<String,Object> result = new HashMap<>();
        Map<String,Object> body = new HashMap<>();
        Map<String,Object> services = new HashMap<>();
        Map<String,Object> dataMap = new HashMap<>();
        dataMap.put("average_windspeed_10min",atmosphereFrame.getAverageWindSpeed10min());
        dataMap.put("average_winddirection_10min",atmosphereFrame.getAverageWindDirection10min());
        dataMap.put("max_windspeed",atmosphereFrame.getMaxWindSpeed());
        dataMap.put("extreme_windspeed",atmosphereFrame.getExtremeWindSpeed());
        dataMap.put("standard_windspeed",atmosphereFrame.getStandardWindSpeed());
        dataMap.put("air_temperature",atmosphereFrame.getAirTemperature());
        dataMap.put("humidity",atmosphereFrame.getHumidity());
        dataMap.put("air_pressure",atmosphereFrame.getAirPressure());
        dataMap.put("precipitation",atmosphereFrame.getPrecipitation());
        dataMap.put("precipitation_intensity",atmosphereFrame.getPrecipitationIntensity());
        dataMap.put("radiation_intensity",atmosphereFrame.getRadiationIntensity());
        services.put("data",dataMap);
        services.put("eventTime",DateUtils.getTimestampDate(atmosphereFrame.getTimeStamp(),"yyyyMMddHHmmss"));
        services.put("serviceId","analog");
        body.put("deviceId",atmosphereFrame.getDeviceIdentification());
        body.put("services",Arrays.asList(services));
        result.put("devices",Arrays.asList(body));
        try {
            processingDatasTopic(atmosphereFrame.getDeviceIdentification(),JSONObject.toJSONString(result));
        } catch (Exception e) {
            return Crc16Utils.replaceCrcStr(atmosphereFrame.getSync().concat("0200").concat(atmosphereFrame.getCmdId()).concat(FrameType.DATA_RESPONSE.getCode()).concat(atmosphereFrame.getPacketType()).concat("01").concat("00").concat("000096"));
        }
        return Crc16Utils.replaceCrcStr(atmosphereFrame.getSync().concat("0200").concat(atmosphereFrame.getCmdId()).concat(FrameType.DATA_RESPONSE.getCode()).concat(atmosphereFrame.getPacketType()).concat("01").concat("FF").concat("000096"));
    }


    /**
     * 处理结冰设备的覆冰数据
     * @param iceCoverFrame
     * @return
     */
    public String processingIceCoverFrame(IceCoverFrame iceCoverFrame){
        Map<String,Object> result = new HashMap<>();
        Map<String,Object> body = new HashMap<>();
        Map<String,Object> services = new HashMap<>();
        Map<String,Object> dataMap = new HashMap<>();
        dataMap.put("temperature",iceCoverFrame.getTemperature());
        dataMap.put("icing_state",iceCoverFrame.getIcingState());
        dataMap.put("icing_thickness",iceCoverFrame.getIcingThickness());
        services.put("data",dataMap);
        services.put("eventTime",DateUtils.getTimestampDate(iceCoverFrame.getTimeStamp(),"yyyyMMddHHmmss"));
        services.put("serviceId","analog");
        body.put("deviceId",iceCoverFrame.getDeviceIdentification());
        body.put("services",Arrays.asList(services));
        result.put("devices",Arrays.asList(body));
        try {
            processingDatasTopic(iceCoverFrame.getDeviceIdentification(),JSONObject.toJSONString(result));
        } catch (Exception e) {
            return Crc16Utils.replaceCrcStr(iceCoverFrame.getSync().concat("0200").concat(iceCoverFrame.getCmdId()).concat(FrameType.DATA_RESPONSE.getCode()).concat(iceCoverFrame.getPacketType()).concat("01").concat("00").concat("000096"));
        }
        return Crc16Utils.replaceCrcStr(iceCoverFrame.getSync().concat("0200").concat(iceCoverFrame.getCmdId()).concat(FrameType.DATA_RESPONSE.getCode()).concat(iceCoverFrame.getPacketType()).concat("01").concat("FF").concat("000096"));
    }

    /**
     * 处理结冰设备的图片数据
     * imageDataFrame
     * @return
     */
    public void processingImageData(ImageDataFrame imageDataFrame){
        List<Integer> subpacketArr = new ArrayList<>();
        Map<Integer,String> subpacketData = new HashMap<>();
        //将子包序列数组和子包sample数据存到redis中
        if(Boolean.TRUE.equals(redisService.hasKey(CacheConstants.IMAGE_SUBPACKET_ARRAY+imageDataFrame.getDeviceIdentification()+"_"+imageDataFrame.getChannelNo()))){
            subpacketArr = redisService.getCacheObject(CacheConstants.IMAGE_SUBPACKET_ARRAY+imageDataFrame.getDeviceIdentification()+"_"+imageDataFrame.getChannelNo());
            subpacketData = redisService.getCacheObject(CacheConstants.IMAGE_PACKET_DATA+imageDataFrame.getDeviceIdentification()+"_"+imageDataFrame.getChannelNo());
        }
        subpacketArr.add(Integer.valueOf(imageDataFrame.getSubpacketNo()));
        subpacketData.put(Integer.valueOf(imageDataFrame.getSubpacketNo()),imageDataFrame.getSample());
        subpacketArr = new ArrayList<Integer>(new TreeSet<Integer>(subpacketArr));
        redisService.setCacheObject(CacheConstants.IMAGE_SUBPACKET_ARRAY+imageDataFrame.getDeviceIdentification()+"_"+imageDataFrame.getChannelNo(),subpacketArr);
        redisService.setCacheObject(CacheConstants.IMAGE_PACKET_DATA+imageDataFrame.getDeviceIdentification()+"_"+imageDataFrame.getChannelNo(),subpacketData);
        redisService.setCacheObject(CacheConstants.IMAGE_FRAME_PACKETNO+imageDataFrame.getDeviceIdentification()+"_"+imageDataFrame.getChannelNo(),Integer.valueOf(imageDataFrame.getPacketNo()));
    }

    public String processingImageSendEnd(ImageRequestEndFrame imageRequestEndFrame) throws Exception {
        Integer packet_no = redisService.getCacheObject(CacheConstants.IMAGE_FRAME_PACKETNO+imageRequestEndFrame.getDeviceIdentification()+"_"+imageRequestEndFrame.getChannelNo());
        if (packet_no == null) return Crc16Utils.replaceCrcStr(imageRequestEndFrame.getSync().concat("0c00").concat(imageRequestEndFrame.getCmdId()).concat(FrameType.VIDEO_CONTROL.getCode()).concat(PacketType.IMAGE_DATA_COMPLEMENT.getCode()).concat("01")
                .concat(imageRequestEndFrame.getChannelNoHex()).concat(imageRequestEndFrame.getPresettingNoHex()).concat("00000000000000000000").concat("000096"));
        List<Integer> subpacketArr = redisService.getCacheObject(CacheConstants.IMAGE_SUBPACKET_ARRAY+imageRequestEndFrame.getDeviceIdentification()+"_"+imageRequestEndFrame.getChannelNo());
        subpacketArr = new ArrayList<Integer>(new TreeSet<Integer>(subpacketArr));
        List<String> complementPackNoList = getComplementPackNo(packet_no,subpacketArr);
        if(null == complementPackNoList || complementPackNoList.size()==0){
            String filePath = mergeImageStr(imageRequestEndFrame.getDeviceIdentification(),imageRequestEndFrame.getChannelNo());
            redisService.deleteObject(CacheConstants.IMAGE_SUBPACKET_ARRAY+imageRequestEndFrame.getDeviceIdentification()+"_"+imageRequestEndFrame.getChannelNo());
            redisService.deleteObject(CacheConstants.IMAGE_PACKET_DATA+imageRequestEndFrame.getDeviceIdentification()+"_"+imageRequestEndFrame.getChannelNo());
            redisService.deleteObject(CacheConstants.IMAGE_FRAME_PACKETNO+imageRequestEndFrame.getDeviceIdentification()+"_"+imageRequestEndFrame.getChannelNo());
            saveImageDataToTdengine(imageRequestEndFrame,filePath);
            return Crc16Utils.replaceCrcStr(imageRequestEndFrame.getSync().concat("0c00").concat(imageRequestEndFrame.getCmdId()).concat(FrameType.VIDEO_CONTROL.getCode()).concat(PacketType.IMAGE_DATA_COMPLEMENT.getCode()).concat("01")
                    .concat(imageRequestEndFrame.getChannelNoHex()).concat(imageRequestEndFrame.getPresettingNoHex()).concat("00000000000000000000").concat("000096"));
        }else{
            String complementPackSum = DataConverter.parseInteger2Hex(complementPackNoList.size(),2);
            String complementPackNo = String.join("",complementPackNoList);
            return Crc16Utils.replaceCrcStr(imageRequestEndFrame.getSync().concat(DataConverter.parseInteger2Hex(complementPackNo.length()+2)).concat(imageRequestEndFrame.getCmdId()).concat(FrameType.VIDEO_CONTROL.getCode()).concat(PacketType.IMAGE_DATA_COMPLEMENT.getCode()).concat("01")
                    .concat(imageRequestEndFrame.getChannelNoHex()).concat(imageRequestEndFrame.getPresettingNoHex()).concat(complementPackSum).concat(complementPackNo).concat("000096"));
        }
    }

    /**
     * 处理远程图像补包数据下发
     * @param data
     * @return
     */

    /**
     * 获取缺少的子包数组
     * @param packetNo
     * @param subpacketList
     * @return
     */
    private List<String> getComplementPackNo(Integer packetNo,List<Integer> subpacketList){
        List<String> complementPackNo = new ArrayList<>();
        // 将现有数加入BitSet
        BitSet bitSet = new BitSet(packetNo);
        for (Integer number : subpacketList) {
            bitSet.set(number);
        }
        // 遍历查找
        for (int i = 1; i <= packetNo; i++) {
            if (!bitSet.get(i)) {
                complementPackNo.add(DataConverter.parseInteger2Hex(i,2));
            }
        }
        return complementPackNo;
    }

    /**
     * 整合图片16进制的16进制数据
     * @param deviceIdentification
     * @param channelNo
     * @return
     */
    private String mergeImageStr(String deviceIdentification,String channelNo){
        List<Integer> subpacketArr = redisService.getCacheObject(CacheConstants.IMAGE_SUBPACKET_ARRAY+deviceIdentification+"_"+channelNo);
        Map<Integer,String> subpacketData = redisService.getCacheObject(CacheConstants.IMAGE_PACKET_DATA+deviceIdentification+"_"+channelNo);
        Collections.sort(subpacketArr);
        String imageHexStr = "";
        for(int i = 1;i<=subpacketArr.size();i++){
            if(null == subpacketData.get(i)) continue;
            imageHexStr = imageHexStr+(subpacketData.get(i));
        }

        //DataConverter.saveToImgFile(imageHexStr.toUpperCase(),"D://image//"+deviceIdentification+"_"+DateUtils.getCurrentDateTime("yyyyMMddHHmmss")+".jpg");
        byte[] bytes = new byte[imageHexStr.length() / 2];
        for (int i = 0; i < bytes.length; i++) {
            bytes[i] = (byte) (0xff & Integer.parseInt(imageHexStr.substring(i * 2, i * 2 + 2).toUpperCase(), 16));
        }
        String path = fastDfsUtils.uploadFile(bytes,bytes.length, "jpg");
        return fdfsPrefix+path;

    }

    /**
     * 保存图片数据到Tdengine
     * @param imageRequestEndFrame
     * @param filePath
     * @throws Exception
     */
    public void saveImageDataToTdengine(ImageRequestEndFrame imageRequestEndFrame, String filePath) throws Exception {
        Map<String,Object> result = new HashMap<>();
        Map<String,Object> body = new HashMap<>();
        Map<String,Object> services = new HashMap<>();
        Map<String,Object> dataMap = new HashMap<>();
        dataMap.put("channel_no",imageRequestEndFrame.getChannelNo());
        dataMap.put("presetting_no",imageRequestEndFrame.getPresettingNo());
        dataMap.put("file_path",filePath);
        services.put("data",dataMap);
        services.put("eventTime",DateUtils.getTimestampDate(imageRequestEndFrame.getTimeStamp(),"yyyyMMddHHmmss"));
        services.put("serviceId","analog");
        body.put("deviceId",imageRequestEndFrame.getDeviceIdentification());
        body.put("services",Arrays.asList(services));
        result.put("devices",Arrays.asList(body));
        processingDatasTopic(imageRequestEndFrame.getDeviceIdentification(),JSONObject.toJSONString(result));
    }
}
